/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2025-2025. All rights reserved.
 * MemCache_Hybrid is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/
#include "mmc_meta_net_client.h"

#include "mmc_msg_base.h"
#include "mmc_msg_client_meta.h"

namespace ock {
namespace mmc {
std::map<std::string, MmcRef<MetaNetClient>> MetaNetClientFactory::instances_;
std::mutex MetaNetClientFactory::instanceMutex_;

MetaNetClient::~MetaNetClient() {}
MetaNetClient::MetaNetClient(const std::string &serverUrl, const std::string& inputName)
    : serverUrl_(serverUrl), name_(inputName)
{}

Result MetaNetClient::Start(const NetEngineOptions &config)
{
    std::lock_guard<std::mutex> guard(mutex_);
    if (started_) {
        MMC_LOG_INFO("MetaNetClient [" << name_ << "] already started");
        return MMC_OK;
    }

    /* init engine */

    NetEnginePtr client = NetEngine::Create();
    MMC_ASSERT_RETURN(client != nullptr, MMC_MALLOC_FAILED);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_PING_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_ALLOC_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_UPDATE_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_UPDATE_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_GET_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_GET_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_REMOVE_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_REMOVE_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BM_REGISTER_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_IS_EXIST_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_IS_EXIST_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BM_UNREGISTER_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_QUERY_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_QUERY_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::ML_BATCH_ALLOC_REQ, nullptr);
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::LM_PING_REQ,
                                      std::bind(&MetaNetClient::HandlePing, this, std::placeholders::_1));
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::LM_META_REPLICATE_REQ,
                                      std::bind(&MetaNetClient::HandleMetaReplicate, this, std::placeholders::_1));
    client->RegRequestReceivedHandler(LOCAL_META_OPCODE_REQ::LM_BLOB_COPY_REQ,
                                      std::bind(&MetaNetClient::HandleBlobCopy, this, std::placeholders::_1));
    client->RegLinkBrokenHandler(std::bind(&MetaNetClient::HandleLinkBroken, this, std::placeholders::_1));
    /* start engine */
    MMC_ASSERT_RETURN(client->Start(config) == MMC_OK, MMC_NOT_STARTED);

    engine_ = client;
    rankId_ = config.rankId;
    started_ = true;
    MMC_LOG_INFO("initialize meta net server success [" << name_ << "]");
    return MMC_OK;
}

void MetaNetClient::Stop()
{
    std::lock_guard<std::mutex> guard(mutex_);
    if (!started_) {
        MMC_LOG_WARN("MetaNetClient has not been started");
        return;
    }
    if (engine_ != nullptr) {
        engine_->Stop();
    }
    started_ = false;
}

Result MetaNetClient::Connect(const std::string &url)
{
    NetEngineOptions options;
    NetEngineOptions::ExtractIpPortFromUrl(url, options);
    MMC_ASSERT_RETURN(engine_ != nullptr, MMC_NOT_INITIALIZED);
    MMC_RETURN_ERROR(engine_->ConnectToPeer(rankId_, options.ip, options.port, link2Index_, false),
                     "MetaNetClient Connect " << url << " failed");
    ip_ = options.ip;
    port_ = options.port;
    return MMC_OK;
}

Result MetaNetClient::HandleMetaReplicate(const NetContextPtr &context)
{
    MetaReplicateRequest req;
    Response resp;
    context->GetRequest<MetaReplicateRequest>(req);
    if (replicateHandler_ != nullptr) {
        resp.ret_ = replicateHandler_(req.ops_, req.keys_, req.blobs_);
    } else {
        MMC_LOG_ERROR("replicateHandler_ is nullptr");
        resp.ret_ = MMC_ERROR;
    }

    return context->Reply(req.msgId, resp);
}

Result MetaNetClient::HandleBlobCopy(const NetContextPtr& context)
{
    BlobCopyRequest req;
    Response resp;
    context->GetRequest<BlobCopyRequest>(req);
    if (blobCopyHandler_ != nullptr) {
        resp.ret_ = blobCopyHandler_(req.srcBlob_, req.dstBlob_);
        if (resp.ret_ != MMC_OK) {
            MMC_LOG_ERROR("blobCopy failed, ret:" << resp.ret_);
        }
    } else {
        MMC_LOG_ERROR("blobCopyHandler_ is nullptr");
        resp.ret_ = MMC_ERROR;
    }
    return context->Reply(req.msgId, resp);
}

Result MetaNetClient::HandlePing(const NetContextPtr &context)
{
    std::string str{(char *)context->Data(), context->DataLen()};
    NetMsgUnpacker unpacker(str);
    PingMsg req;
    req.Deserialize(unpacker);
    MMC_LOG_INFO("HandlePing num " << req.num);

    NetMsgPacker packer;
    PingMsg recv;
    recv.Serialize(packer);
    std::string serializedData = packer.String();
    uint32_t retSize = serializedData.length();
    return context->Reply(req.msgId, const_cast<char *>(serializedData.c_str()), retSize);
}

Result MetaNetClient::HandleLinkBroken(const NetLinkPtr &link)
{
    MMC_LOG_INFO(name_ << " link broken");
    MMC_ASSERT_RETURN(engine_ != nullptr, MMC_NOT_INITIALIZED);
    for (uint32_t count = 0; count < retryCount_; count++) {
        Result ret = engine_->ConnectToPeer(rankId_, ip_, port_, link2Index_, false);
        if (ret != MMC_OK) {
            MMC_LOG_ERROR("MetaNetClient Connect " << ip_ << ", port " << port_ << " failed");
        } else {
            if (retryHandler_ != nullptr) {
                MMC_LOG_INFO("call retry handler when reconnect to " << ip_ << ", port " << port_);
                return retryHandler_();
            }
            return MMC_OK;
        }
        sleep(2ULL);
    }
    return MMC_ERROR;
}
}
}